package com.gitee.source

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.source.{RichParallelSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment, _}

object CustomerSourceWithRichParallel {
  def main(args: Array[String]): Unit = {
    val senv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val data: DataStream[Long] = senv.addSource(new MyRichParallel()).setParallelism(3)
    data.print()
    senv.execute()
  }

}

class MyRichParallel extends RichParallelSourceFunction[Long] () {

  override def close(): Unit = {
//    关闭连接
  }

  override def open(parameters: Configuration): Unit = {
//    开启连接
  }

  var currentNum:Long = 1
  var isRunning:Boolean = true//表示是否正在执行
  override def cancel(): Unit = {
    isRunning = false
  }

  override def run(ctx: SourceFunction.SourceContext[Long]): Unit = {
    while(isRunning){
      //使用ctx将数据生成出去
      ctx.collect(currentNum)
      currentNum+=1
      Thread.sleep(1000)
    }
  }
  //TODO 取消和停止（流作业）的区别如下：
  //cancel()调用，立即调用作业算子的 cancel() 方法，以尽快取消它们。如果算子在接到 cancel() 调用后没有停止，Flink 将开始定期中断算子线程的执行，直到所有算子停止为止。
  //stop()调用，是更优雅的停止正在运行流作业的方式。
  // stop() 仅适用于 Source 实现了 StoppableFunction 接口的作业。
  // 当用户请求停止作业时，作业的所有 Source 都将接收 stop() 方法调用。
  // 直到所有 Source 正常关闭时，作业才会正常结束。这种方式，使作业正常处理完所有作业。
}
